/** 
 * Project Name: farm-core 
 * File Name: SmartPool.java 
 * Package Name: com.bugull.farm.core.frame 
 * Date: 2017年3月8日下午6:04:21 
 * Copyright (c) 2017, hadlinks All Rights Reserved. 
 * 
 */
package com.bugull.farm.core.frame.pool;

import java.io.Closeable;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import com.codahale.metrics.Gauge;
import com.codahale.metrics.MetricRegistry;
import io.netty.util.concurrent.DefaultThreadFactory;

/** 
 * ClassName: AbstractSmartPool 
 * Function: 自定义线程池模型的抽象类，可以自定义任务分配算法接口，任务失败处理接口.  
 * date: 2017年3月8日 下午6:04:21  
 * 
 * @author songwei (songw@hadlinks.com)
 * @version 
 * @since JDK 1.8 
 */
public abstract class AbstractSmartPool implements Closeable{

	//线程池模型
	private final List<ThreadPoolExecutor> threadPools;
	
	//每个线程对应的消息队列
	private final List<LinkedBlockingQueue<Runnable>> queues;
	
	//线程池大小
    private final int size;

    /**
     * 构造函数，加入一个Metric度量器，监控线程中队列的一些状态
     * @param size
     * @param metricRegistry
     */
    AbstractSmartPool(int size, MetricRegistry metricRegistry) {
        this.size = size;
        this.queues = IntStream.range(0, size).mapToObj(i -> new LinkedBlockingQueue<Runnable>()).collect(Collectors.toList());
        List<ThreadPoolExecutor> list = queues.stream()
                .map(q -> new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, q, new DefaultThreadFactory("Bugu")))
                .collect(Collectors.toList());
        metricRegistry.register(MetricRegistry.name(ProjectionMetrics.class, "queue"), (Gauge<Double>) this::averageQueueLength);
        //这里使用CopyOnWriteArrayList是为了保证访问threadPools里面元素时是线程安全的
        this.threadPools = new CopyOnWriteArrayList<>(list);
    }
    
    /**
     * 计算队列平均长度
     * @return
     */
    private double averageQueueLength() {
        double totalLength =
            queues.stream().mapToDouble(LinkedBlockingQueue::size).sum();
        return totalLength / queues.size();
    }

    /**
     * 线程池关闭方法
     */
    @Override
    public void close() throws IOException {
        threadPools.forEach(ExecutorService::shutdown);
    }

    /**
     * 提交任务
     * @param callable
     */
    public void submit(BuguCallable<Boolean> callable) {
    	System.out.println("提交一个任务");
        final int threadIdx = distribution(callable);
        final ExecutorService executor = threadPools.get(threadIdx);
        executor.submit(new BuguFutureTask(callable));
    }

    /**
     * 提交任务
     * @param callable
     */
    public void execute(BuguCallable<Boolean> callable) {
        System.out.println("执行一个任务");
        final int threadIdx = distribution(callable);
        final ExecutorService executor = threadPools.get(threadIdx);
        executor.execute(new BuguFutureTask(callable));
    }

    public abstract int distribution(BuguCallable<Boolean> callable);
    
	public int getSize() {
		return size;
	}
}
